🎖️GitЯра🎖️
Commit 84bc6fd48ed2173aa688b9c66e7c7616b3fedf04
Parents : 06aee11
Author : Jeremiah K <17190268+jeremiah-k@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-08-15T18:56:22Z
Committer : GitHub <noreply@github.com>
Date : 2026-08-15T18:56:22Z
fix(admin): retain session refresh across multi-hop latency (#6718)
Changes
3 files changed, 178 insertions(+), 25 deletions(-)
Diff
diff --git a/core/domain/README.md b/core/domain/README.md
index bf2ba4a66e..596653c7f7 100644
--- a/core/domain/README.md
+++ b/core/domain/README.md
@@ -50,7 +50,7 @@ Ensures a per-node remote-admin passkey session exists before entering the remot
sealed interface EnsureSessionResult {
data object AlreadyActive : EnsureSessionResult // passkey already fresh
data object Refreshed : EnsureSessionResult // metadata response arrived
- data object Timeout : EnsureSessionResult // no response within 10 s
+ data object Timeout : EnsureSessionResult // no response within 30 s
data object Disconnected : EnsureSessionResult // radio not connected
}
```
diff --git a/core/domain/src/commonMain/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCase.kt b/core/domain/src/commonMain/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCase.kt
index 03775644ce..9df3d09d93 100644
--- a/core/domain/src/commonMain/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCase.kt
+++ b/core/domain/src/commonMain/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCase.kt
@@ -19,11 +19,15 @@ package org.meshtastic.core.domain.usecase.session
import co.touchlab.kermit.Logger
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.selects.select
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeoutOrNull
import org.koin.core.annotation.Single
import org.meshtastic.core.common.di.ServiceScope
@@ -47,6 +51,8 @@ import kotlin.time.Duration.Companion.seconds
* blast two metadata requests at the radio.
* - The refresh-flow subscription is established **before** the metadata request is dispatched to avoid losing the
* response on the inherently raceful `MutableSharedFlow`.
+ * - A connection departure completes the shared ensure as [EnsureSessionResult.Disconnected], so a later connection
+ * never inherits an old request or UX deadline.
* - The `withTimeoutOrNull` is a UX deadline only — late responses still update the durable `SessionStatus` flow that
* the UI observes, so a "Timeout" outcome here can self-heal in the chip without re-tapping.
*/
@@ -57,7 +63,7 @@ open class EnsureRemoteAdminSessionUseCase(
private val serviceRepository: ServiceRepository,
private val serviceScope: ServiceScope,
) {
- private val mutex = Mutex()
+ private val inFlightMutex = Mutex()
private val inFlight = mutableMapOf<Int, Deferred<EnsureSessionResult>>()
@Suppress("ReturnCount")
@@ -70,42 +76,73 @@ open class EnsureRemoteAdminSessionUseCase(
}
val deferred =
- mutex.withLock {
- inFlight[destNum]
- ?: serviceScope
- .async(start = CoroutineStart.LAZY) { runEnsure(destNum) }
- .also { inFlight[destNum] = it }
+ inFlightMutex.withLock {
+ inFlight[destNum]?.takeIf { !it.isCompleted }
+ ?: run {
+ lateinit var newDeferred: Deferred<EnsureSessionResult>
+ newDeferred =
+ serviceScope.async(start = CoroutineStart.LAZY) {
+ try {
+ runEnsure(destNum)
+ } finally {
+ // Cleanup belongs to the shared Deferred itself. NonCancellable guarantees that a
+ // service-scope cancellation cannot strand a completed entry in the dedupe map.
+ withContext(NonCancellable) {
+ inFlightMutex.withLock {
+ if (inFlight[destNum] === newDeferred) inFlight.remove(destNum)
+ }
+ }
+ }
+ }
+ // Register before the lazy deferred starts. The identity check above prevents an old completion
+ // from removing a newer ensure for the same node.
+ inFlight[destNum] = newDeferred
+ // The service scope, not the first awaiting caller, owns dispatch once the entry is visible.
+ // A lazy child can already be cancelled when its parent scope is shutting down; in that case
+ // its body never runs, so remove the entry here instead of relying on the body-level finally.
+ if (!newDeferred.start() && inFlight[destNum] === newDeferred) inFlight.remove(destNum)
+ newDeferred
+ }
}
- return try {
- deferred.await()
- } finally {
- mutex.withLock { if (inFlight[destNum] === deferred) inFlight.remove(destNum) }
- }
+ return deferred.await()
}
private suspend fun runEnsure(destNum: Int): EnsureSessionResult {
Logger.d { "EnsureRemoteAdminSession dispatching metadata request to $destNum" }
return withTimeoutOrNull(UX_TIMEOUT) {
- // Subscribe BEFORE dispatching so we don't miss the refresh emission.
- val refreshed =
- serviceScope.async(start = CoroutineStart.UNDISPATCHED) {
- sessionManager.sessionRefreshFlow.filter { it == destNum }.first()
+ coroutineScope {
+ // Subscribe to both terminal events before dispatch so neither a synchronous response nor a
+ // concurrent connection departure can be missed. UNDISPATCHED is deliberate: these children do no
+ // work before suspending in first(), so starting inline only establishes the subscriptions.
+ val refreshed =
+ async(start = CoroutineStart.UNDISPATCHED) {
+ sessionManager.sessionRefreshFlow.filter { it == destNum }.first()
+ }
+ val disconnected =
+ async(start = CoroutineStart.UNDISPATCHED) {
+ serviceRepository.connectionState.filter { it != ConnectionState.Connected }.first()
+ }
+ try {
+ if (disconnected.isCompleted) return@coroutineScope EnsureSessionResult.Disconnected
+ radioController.refreshMetadata(destNum)
+ select<EnsureSessionResult> {
+ disconnected.onAwait { EnsureSessionResult.Disconnected }
+ refreshed.onAwait { EnsureSessionResult.Refreshed }
+ }
+ } finally {
+ disconnected.cancel()
+ refreshed.cancel()
}
- try {
- radioController.refreshMetadata(destNum)
- refreshed.await()
- EnsureSessionResult.Refreshed
- } finally {
- refreshed.cancel()
}
} ?: EnsureSessionResult.Timeout
}
companion object {
/**
- * UX deadline for surfacing a result to the user. The metadata request keeps flying after this — late responses
- * still update the durable `SessionStatus` flow.
+ * UX deadline for surfacing a result to the user. Multi-hop remote-admin responses can legitimately take well
+ * over ten seconds; the metadata request keeps flying after this, and late responses still update the durable
+ * `SessionStatus` flow.
*/
- val UX_TIMEOUT = 10.seconds
+ val UX_TIMEOUT = 30.seconds
}
}
diff --git a/core/domain/src/commonTest/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCaseTest.kt b/core/domain/src/commonTest/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCaseTest.kt
index 2cc9c13a62..f7d865118a 100644
--- a/core/domain/src/commonTest/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCaseTest.kt
+++ b/core/domain/src/commonTest/kotlin/org/meshtastic/core/domain/usecase/session/EnsureRemoteAdminSessionUseCaseTest.kt
@@ -25,12 +25,15 @@ import dev.mokkery.matcher.any
import dev.mokkery.mock
import dev.mokkery.verifySuspend
import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import okio.ByteString
import org.meshtastic.core.common.di.asServiceScope
@@ -42,6 +45,7 @@ import org.meshtastic.core.repository.SessionManager
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.time.Clock
+import kotlin.time.Duration.Companion.seconds
@OptIn(ExperimentalCoroutinesApi::class)
class EnsureRemoteAdminSessionUseCaseTest {
@@ -116,6 +120,28 @@ class EnsureRemoteAdminSessionUseCaseTest {
verifySuspend { controller.refreshMetadata(destNum) }
}
+ @Test
+ fun `accepts a valid refresh arriving after the former ten second deadline`() = runTest {
+ val refresh = MutableSharedFlow<Int>(extraBufferCapacity = 8)
+ val sessionManager = stubSessionManager(refreshFlow = refresh)
+ val controller = mock<RadioController>(MockMode.autofill)
+ everySuspend { controller.refreshMetadata(any()) } returns Unit
+ val useCase =
+ EnsureRemoteAdminSessionUseCase(sessionManager, controller, connectedRepo(), this.asServiceScope())
+
+ var observed: EnsureSessionResult? = null
+ val job = launch { observed = useCase(destNum) }
+ runCurrent()
+ // Keep the emission late while deriving the boundary from the production UX deadline.
+ advanceTimeBy(EnsureRemoteAdminSessionUseCase.UX_TIMEOUT.inWholeMilliseconds - 1.seconds.inWholeMilliseconds)
+ refresh.emit(destNum)
+ advanceUntilIdle()
+ job.join()
+
+ assertEquals(EnsureSessionResult.Refreshed, observed)
+ verifySuspend { controller.refreshMetadata(destNum) }
+ }
+
@Test
fun `returns Timeout when no refresh arrives within deadline`() = runTest {
val refresh = MutableSharedFlow<Int>(extraBufferCapacity = 8)
@@ -134,4 +160,94 @@ class EnsureRemoteAdminSessionUseCaseTest {
assertEquals(EnsureSessionResult.Timeout, observed)
}
+
+ @Test
+ fun `canceling one caller keeps the shared ensure alive for another caller`() = runTest {
+ val refresh = MutableSharedFlow<Int>(extraBufferCapacity = 8)
+ val sessionManager = stubSessionManager(refreshFlow = refresh)
+ val controller = mock<RadioController>(MockMode.autofill)
+ var dispatches = 0
+ everySuspend { controller.refreshMetadata(any()) } calls
+ {
+ dispatches++
+ Unit
+ }
+ val useCase =
+ EnsureRemoteAdminSessionUseCase(sessionManager, controller, connectedRepo(), this.asServiceScope())
+
+ val firstCaller = launch { useCase(destNum) }
+ runCurrent()
+ val secondCaller = async { useCase(destNum) }
+ runCurrent()
+
+ firstCaller.cancelAndJoin()
+ assertEquals(1, dispatches)
+ refresh.emit(destNum)
+ runCurrent()
+
+ assertEquals(EnsureSessionResult.Refreshed, secondCaller.await())
+ assertEquals(1, dispatches)
+ }
+
+ @Test
+ fun `completed ensure is not reused by a later caller`() = runTest {
+ val refresh = MutableSharedFlow<Int>(extraBufferCapacity = 8)
+ val sessionManager = stubSessionManager(refreshFlow = refresh)
+ val controller = mock<RadioController>(MockMode.autofill)
+ var dispatches = 0
+ everySuspend { controller.refreshMetadata(any()) } calls
+ {
+ dispatches++
+ Unit
+ }
+ val useCase =
+ EnsureRemoteAdminSessionUseCase(sessionManager, controller, connectedRepo(), this.asServiceScope())
+
+ val first = async { useCase(destNum) }
+ runCurrent()
+ refresh.emit(destNum)
+ runCurrent()
+ assertEquals(EnsureSessionResult.Refreshed, first.await())
+
+ val second = async { useCase(destNum) }
+ runCurrent()
+ assertEquals(2, dispatches)
+ refresh.emit(destNum)
+ runCurrent()
+
+ assertEquals(EnsureSessionResult.Refreshed, second.await())
+ }
+
+ @Test
+ fun `departed ensure is not reused after reconnect`() = runTest {
+ val refresh = MutableSharedFlow<Int>(extraBufferCapacity = 8)
+ val sessionManager = stubSessionManager(refreshFlow = refresh)
+ val controller = mock<RadioController>(MockMode.autofill)
+ var dispatches = 0
+ everySuspend { controller.refreshMetadata(any()) } calls
+ {
+ dispatches++
+ Unit
+ }
+ val connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Connected)
+ val repository = mock<ServiceRepository>(MockMode.autofill)
+ every { repository.connectionState } returns connectionState
+ val useCase = EnsureRemoteAdminSessionUseCase(sessionManager, controller, repository, this.asServiceScope())
+
+ val departed = async { useCase(destNum) }
+ runCurrent()
+ assertEquals(1, dispatches)
+ connectionState.value = ConnectionState.Disconnected
+ runCurrent()
+ assertEquals(EnsureSessionResult.Disconnected, departed.await())
+
+ connectionState.value = ConnectionState.Connected
+ val reconnected = async { useCase(destNum) }
+ runCurrent()
+ assertEquals(2, dispatches, "Reconnect must dispatch a new metadata request")
+ refresh.emit(destNum)
+ runCurrent()
+
+ assertEquals(EnsureSessionResult.Refreshed, reconnected.await())
+ }
}
Served by rngit 1.5.0 - Generated in 0.07s